package voldemort.store.cachestore.voldeimpl;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import voldemort.server.VoldemortConfig;
import voldemort.store.StorageConfiguration;
import voldemort.store.StorageEngine;
import voldemort.store.StoreDefinition;
import voldemort.store.cachestore.BlockSize;
import voldemort.store.cachestore.impl.Purge;
import voldemort.utils.ByteArray;
import voldemort.utils.Props;
import voldemort.utils.ReflectUtils;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Created by IntelliJ IDEA.
 * User: mhsieh
 * Date: 1/26/11
 * Time: 9:37 AM
 * To change this template use File | Settings | File Templates.
 */
public class VoldeStoreConfiguration implements StorageConfiguration {
  private static Log logger = LogFactory.getLog(VoldeStoreConfiguration.class);

    public static final String TYPE_NAME = "voldestore";
    private VoldemortConfig config;
    private final ConcurrentMap<String, VoldeStore> stores = new ConcurrentHashMap<String, VoldeStore>();
    private final Object lock = new Object();
    public static final String DELAY_WRITE ="delay.write";
    public static final String BLOCK_SIZE  = "block.size.class";
    public static final String PURGE_INTERVAL = "purge.interval";
    public static final String PURGE_CLASS ="purge.class";
    public static final String TRIGGER = "trigger.size";
    public static final String PACK_INTERVAL = "pack.interval";
    public static final String WRITE_THREAD="write.thread";
    public static final String FILE_MODE ="file.mode";




    public VoldeStoreConfiguration(VoldemortConfig config) {
        this.config = config ;
    }

   public StorageEngine<ByteArray, byte[], byte[]> getStore(StoreDefinition storeDef){
        return getStore( storeDef.getName() );
    }

    public StorageEngine<ByteArray, byte[], byte[]> getStore(String name) {

        VoldeStore vs = stores.get(name);
        if ( vs != null ) return vs;
        // else, create a new store
        synchronized(lock) {
            // double check it again
            vs = stores.get(name);
            // create a new one, when it is not existing
            if ( vs == null ) {
                Props props = loadProp(config);
                String blockSizeName = props.getString(BLOCK_SIZE, null);
                boolean delay = props.getBoolean(DELAY_WRITE, false);
                logger.info("delay.write "+delay);
                int mode = props.getInt(name+"."+FILE_MODE, 0);
                logger.info(name+"."+FILE_MODE+" "+mode);
                if ( blockSizeName != null ) {
                    logger.info("using BlockSize interface of "+ blockSizeName );
                    Class blockSizeClass = ReflectUtils.loadClass(blockSizeName);
                    try {
                        BlockSize blockSize = (BlockSize) blockSizeClass.newInstance() ;
                        vs = new VoldeStore<ByteArray, byte[], byte[]>(name, blockSize, config.getDataDirectory(),
                                (short) config.getNodeId() , delay, mode);
                    } catch ( Exception ex) {
                        logger.warn("Not able to create BlockSize interface using default "+ ex.getMessage());
                        vs = new VoldeStore<ByteArray, byte[], Byte[]>(name, null, config.getDataDirectory(),
                                (short) config.getNodeId(), delay, mode);
                    }
                }
                else {
                    vs = new VoldeStore<ByteArray, byte[], byte[]>(name, null, config.getDataDirectory(),
                        (short) config.getNodeId(), delay);
                }
                // passing props to Store;
                vs.setProp( props);
                VoldeStore vstore= stores.putIfAbsent(name, vs);
                //check it agin
                if ( vstore != null ) vs = vstore ;
                else {
                    if ( delay ) {
                       // default 3 write thread
                       int no = props.getInt(WRITE_THREAD, 3);
                       logger.info("Start write back thread "+no);
                       vs.startWriteThread( no);
                    }
                    // trigger size in mb
                    long trigger = props.getLong(TRIGGER, 0);
                    if ( trigger > 0 ) {
                        logger.info("Purge trigger "+trigger );
                        vs.getCacheStore().setTrigger( trigger);
                    }
                   // start pack Thread
                    long packPeriod = props.getLong("pack.interval", 0);
                    logger.info("start pack data thread at hour interval "+ packPeriod );
                    vs.startPackThread( packPeriod * 3600 * 1000);
                    // pass statList ot cacheStore
                    logger.info("Pass state list to CacheStore");
                    vs.passStatList();
                    try {
                        String purgeClassName = props.getString(name+"."+PURGE_CLASS, null);
                        if ( purgeClassName != null ) {
                            logger.info("using purge interface of "+ purgeClassName );
                            Class purgeClass = ReflectUtils.loadClass( purgeClassName);
                            Purge purge = (Purge) purgeClass.newInstance() ;
                            //default is  0, not start backup thread
                            long period = props.getLong(PURGE_INTERVAL, 0 );
                            vs.startTask( period * 3600 * 1000, purge);
                        }
                    }catch (Exception ex) {
                        logger.error( "fail to start purge thread due to"+ex.getMessage(), ex);
                        //logger.info("fail to s");
                    }
                    // start flush thread
                    int period = props.getInt("flush.period", 0);
                    if ( period > 0 && mode == 0 ) {
                        logger.info("start flush task in minutes "+period );
                        vs.startFlushThread( period);
                    }
                    //}
                }
            }
            return vs;
        } //lock
    }

    public String getType() {
        return TYPE_NAME;
    }

    public void close() {
        Collection<VoldeStore> allStores = stores.values();
        for ( VoldeStore each : allStores ) {
            logger.info("Close store " + each.getName());
            each.close();
        }
    }

    public void update(StoreDefinition storeDef) {
        logger.warn("update storeDef is not supported "+storeDef.getName() );
    }


    public static Props loadProp(VoldemortConfig config) {
        String voldemortHome = config.getVoldemortHome();
        if(voldemortHome == null)
            throw new RuntimeException("No environment variable "
                                             + VoldemortConfig.VOLDEMORT_HOME_VAR_NAME
                                             + " has been defined, set it!");

        String propertiesFile = voldemortHome + File.separator + "config" + File.separator
                                + "server.properties";
        Props properties = null;
        try {
            properties = new Props(new File(propertiesFile));
            properties.put("voldemort.home", voldemortHome);
            String delayWrite = System.getProperty(DELAY_WRITE);
            if ( delayWrite != null && delayWrite.equals("true")) properties.put(DELAY_WRITE, "true");
            String blockSize = System.getProperty(BLOCK_SIZE);
            if ( blockSize != null ) properties.put(BLOCK_SIZE, blockSize);

            return properties;
        } catch(IOException e) {
            throw new RuntimeException(e);
        }
    }

}

